package cn.tecnova.cd

import cn.tecnova.utils.ConfigHandler
import com.alibaba.fastjson.{JSON, JSONObject}
import com.google.gson.Gson
import org.apache.commons.lang3.StringUtils
import org.apache.commons.pool2.impl.GenericObjectPoolConfig
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{SQLContext, SaveMode}
import org.elasticsearch.spark.rdd.EsSpark
import redis.clients.jedis.JedisPool

/**
  * description:
  * Rabcheng
  * Date:2019/6/20 15:26
  **/
object Base2mysql {


  def main(args: Array[String]): Unit = {


    val conf = new SparkConf()
      .setAppName("estest")
    //      .setMaster("local[*]")
    conf.set("es.nodes", "172.17.22.16")
    conf.set("es.port", "9600")
    val sc = new SparkContext(conf)

    val sQLContext = new SQLContext(sc)

    val rdd: RDD[(String, String)] = EsSpark.esJsonRDD(sc, "base/zhangguoao")

    /*rdd.foreachPartition(iter => {
//      val pool = new JedisPool("172.17.22.18", 8014)
      val pool = new JedisPool(new GenericObjectPoolConfig, "172.17.22.18", 8014, 1000, "123456", 0)
      val jedis = pool.getResource
      iter.foreach(tp => {
        println(tp._2)
        jedis.rpush("base", tp._2)
        println("写入成功...")
      })

    })*/

    import sQLContext.implicits._

    val res: RDD[Base] = rdd.map(tp => {

      val base = JSON.parseObject(tp._2,classOf[Base])
      base
    })
    res.toDF().write.mode(SaveMode.Append).jdbc("jdbc:mysql://www.slave4.com:3306/tec_base2", "base", ConfigHandler.props2)


    sc.stop()


  }

}


case class Base(

                 var id: Int,
                 var keyNo: String,
                 var name: String,
                 var no: String,
                 var belongOrg: String,
                 var operName: String,
                 var startDate: String,
                 var endDate: String,
                 var status: String,
                 var updatedDate: String,
                 var shortStatus: String,
                 var creditCode: String,
                 var registCapi: String,
                 var econKind: String,
                 var industryCode: String,
                 var subIndustryCode: String,
                 var industry: String,
                 var subIndustry: String,
                 var scope: String,
                 var contactNumber: String,
                 var email: String,
                 var webSite: String,
                 var imageUrl: String,
                 var orgNo: String,
                 var englishName: String,
                 var product: String,
                 var `type`: String,
                 var originalName: String,
                 var hitField: String,
                 var hitValue: String,
                 var address: String,
                 var provinceCode: String,
                 var cityCode: String,
                 var countyCode: String,
                 var province: String,
                 var city: String,
                 var county: String,
                 var registCapiNum: String

               )
